package com.etl.dataflow.executor.core;

import lombok.SneakyThrows;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author dx
 * @since 2022/4/13
 */
public class MemoryChannel<E> {

    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final int maxSize;
    private final Object[] array;
    private final AtomicInteger front = new AtomicInteger(0);
    private final AtomicInteger rear = new AtomicInteger(0);

    public MemoryChannel() {
        this(10);
    }

    public MemoryChannel(int bufferSize) {
        this.maxSize = bufferSize;
        array = new Object[bufferSize + 1];
    }

    @SneakyThrows
    public void add(E elem) {
        lock.lock();
        try {
            while (isFull()) {
                notFull.await();
            }
            array[rear.get()] = elem;
            rear.set(rear.incrementAndGet() % maxSize);
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    @SneakyThrows
    public E take() {
        lock.lock();
        try {
            while (isEmpty()) {
                notEmpty.await();
            }
            @SuppressWarnings("unchecked")
            E e = (E) array[front.get()];
            array[front.get()] = null;
            front.set(front.incrementAndGet() % maxSize);
            notFull.signalAll();
            return e;
        } finally {
            lock.unlock();
        }
    }

    public boolean isEmpty() {
        return front.get() == rear.get();
    }

    public void clear() {
        Arrays.fill(this.array, null);
        rear.set(0);
        front.set(0);
    }

    public int size() {
        return (this.rear.get() - this.front.get()) % maxSize;
    }

    private boolean isFull() {
        return (rear.get() + 1) % maxSize == front.get();
    }
}
